Add shared base for KIP-1356 IQv2 headers-aware integration tests by Jess668 · Pull Request #4461 · confluentinc/schema-registry

Jess668 · 2026-07-28T20:40:54Z

What

Introduces HeadersIQv2IntegrationTestBase: shared cluster/serde/streams-lifecycle and output-topic barrier helpers, plus sendAndCapture (captures the produced schema-id GUID) and assertSchemaIdHeaders (17-byte MAGIC_BYTE_V1 format + byte equality vs the produced GUID) for the KIP-1356 headers-aware IQv2 tests.

Checklist

Please answer the questions with Y, N or N/A if not applicable.

[ ] Contains customer facing changes? Including API/behavior changes
[ ] Is this change gated behind config(s)?
- List the config(s) needed to be set to enable this change
[ ] Did you add sufficient unit test and/or integration test coverage for this PR?
- If not, please explain why it is not required
[ ] Does this change require modifying existing system tests or adding new system tests?
- If so, include tracking information for the system test changes
[ ] Must this be released together with other change(s), either in this repo or another one?
- If so, please include the link(s) to the changes that must be released together

References

JIRA:

Test & Review

Open questions / Follow-ups

Introduces HeadersIQv2IntegrationTestBase: shared cluster/serde/streams-lifecycle and output-topic barrier helpers, plus sendAndCapture (captures the produced schema-id GUID) and assertSchemaIdHeaders (17-byte MAGIC_BYTE_V1 format + byte equality vs the produced GUID) for the KIP-1356 headers-aware IQv2 tests.

confluent-cla-assistant · 2026-07-28T20:41:06Z

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

Alieh Saeedi (aliehsaeedii)

Thanks Jess668. Made the 1st pass.

Alieh Saeedi (aliehsaeedii) · 2026-07-29T08:56:09Z

+public abstract class HeadersIQv2IntegrationTestBase extends ClusterTestHarness {
+
+    /** GUID bytes the producer wrote for the key / value schema, captured on send (see below). */
+    private byte[] expectedKeyGuid;


These hold only the last captured GUID, so assertSchemaIdHeaders checks every record against the most recent send. That is correct only because every record in these tests shares one key and one value schema — a subclass with two schemas would compare against the wrong GUID. Returning the captured bytes from sendAndCapture and passing them in would make it per-record, as the javadoc claims.

Alieh Saeedi (aliehsaeedii) · 2026-07-29T08:56:09Z

+     */
+    protected void assertSchemaIdHeaders(Headers headers, String context) {
+        byte[] keyBytes = assertGuidHeader(headers, SchemaId.KEY_SCHEMA_ID_HEADER, "Key", context);
+        if (expectedKeyGuid != null) {


If a subclass asserts headers on a record it did not produce through sendAndCapture, both byte-equality checks are skipped and this passes as a format-only check with no signal. Add assertNotNull(expectedKeyGuid, ...) so a missed capture fails.

Alieh Saeedi (aliehsaeedii) · 2026-07-29T08:56:09Z

+        boolean running = false;
+        try {
+            running = startedLatch.await(timeoutSeconds, TimeUnit.SECONDS);
+            assertTrue(running, "KafkaStreams should reach RUNNING state");


A startup failure never counts the latch down, so this waits out the full timeout (90s in the restore tests) and reports only "should reach RUNNING state", leaving the real exception in the log. TimestampedKeyValueStoreWithHeadersIntegrationTest keeps the last observed state in an AtomicReference and puts it in the message — good to carrying that over, plus counting down on terminal states so it fails fast.

Alieh Saeedi (aliehsaeedii) · 2026-07-29T08:56:09Z

+        int timeoutSeconds, Integer commitIntervalMs) throws Exception {
+        CountDownLatch startedLatch = new CountDownLatch(1);
+        KafkaStreams streams = new KafkaStreams(topology, createStreamsProps(appId, commitIntervalMs));
+        streams.cleanUp();


cleanUp() and start() are outside the try, so if either throws, streams is never closed and the caller never gets the reference to close it either — the state-directory lock and any started StreamThreads leak for the rest of the fork JVM. Move both inside the try.

Alieh Saeedi (aliehsaeedii) · 2026-07-29T08:56:09Z

+
+    protected void closeStreams(KafkaStreams streams) {
+        if (streams != null) {
+            streams.close(Duration.ofSeconds(10));


close(Duration) returns false on timeout instead of throwing, and that result is dropped. The restore tests close and immediately restart on the same application.id and state dir, so a timed-out shutdown shows up later as a confusing cleanUp() failure — assert the return value here.

Alieh Saeedi (aliehsaeedii) · 2026-07-29T08:56:09Z

+                }
+            }
+        }
+        assertEquals(expectedCount, results.size(),


The loop stops at >= expectedCount but this asserts exact equality, so a single at-least-once duplicate fails the barrier instead of passing it. Every caller discards the returned list, so assertTrue(results.size() >= expectedCount, ...) is enough.

Alieh Saeedi (aliehsaeedii) · 2026-07-29T08:56:09Z

+    protected void createTopicsWithPartitions(int numPartitions, String... topicNames)
+        throws Exception {
+        Properties adminProps = new Properties();
+        adminProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);


This is an AdminClient, so AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG is the right constant — same value, and it's what the rest of the repo uses.

Alieh Saeedi (aliehsaeedii) · 2026-07-29T08:56:09Z

+        try (KafkaProducer<GenericRecord, GenericRecord> producer =
+                 new KafkaProducer<>(createProducerProps())) {
+            sendAndCapture(producer, new ProducerRecord<>(topic, null, timestamp, key, value));
+            producer.flush();


sendAndCapture already blocks on send().get(), so this flush (and the one in produceAll) is a no-op — drop it.

Alieh Saeedi (aliehsaeedii) · 2026-07-29T08:56:09Z

+        }
+    }
+
+    protected void produce(String topic, GenericRecord key, GenericRecord value, long timestamp)


The KV-store test in #4462 reimplements this instead of calling it — its produceTombstone is just produce(topic, key, null, ts). Can it reuse these two?

Yes, #4462 should call the base's produce/produceAll and make produceTombstone a one-liner. I'll do that in #4462 — no change here.

Alieh Saeedi (aliehsaeedii) · 2026-07-29T08:56:09Z

+    // Header-mode serdes and client configs
+    // ---------------------------------------------------------------------------------------------
+
+    protected GenericAvroSerde createKeySerde() {


Each serde owns a CachedSchemaRegistryClient and its HTTP connection pool, and nothing closes them — Streams does not close serdes passed via Consumed.with/Produced.with/StoreBuilder. Every other client here is in a try-with-resources; these leak one pool each per topology build.

- Use AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG for the admin client - Fail fast in startStreamsAndAwaitRunning on terminal states and report the last observed state; move cleanUp()/start() inside the try - Assert KafkaStreams.close(Duration) did not time out - Accept at-least-once duplicates in the output-topic barrier - Drop redundant producer.flush() after blocking send().get() - Capture the produced schema-id GUIDs per record; a null/missing capture now fails rather than silently degrading to a format-only check - Close created serdes in teardown

Jess668 requested a review from a team as a code owner July 28, 2026 20:40

Alieh Saeedi (aliehsaeedii) reviewed Jul 29, 2026

View reviewed changes

Uh oh!

Conversation

Jess668 commented Jul 28, 2026

What

Checklist

References

Test & Review

Open questions / Follow-ups

Uh oh!

confluent-cla-assistant Bot commented Jul 28, 2026

Uh oh!

Alieh Saeedi (aliehsaeedii) left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants